package com.baiyan.kafka;


import com.alibaba.fastjson.JSON;
import com.baiyan.entity.TOrderExt;
import com.baiyan.model.OrderDTO;
import com.baiyan.service.ITOrderExtService;
import com.baiyan.service.OrderService;
import com.baiyan.util.GsonUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.google.gson.JsonArray;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.AbstractConsumerSeekAware;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.lang.reflect.InvocationTargetException;
import java.util.List;

/**
 * 订单消费者
 *
 * @author baiyan
 * @date 2022/01/19
 */
@Component
@Slf4j
@ConfigurationProperties(prefix = "kafka.order")
@Data
@EqualsAndHashCode(callSuper = false)
public class OrderKafkaListener extends AbstractConsumerSeekAware {

    @Autowired
    private OrderService orderService;
    @Autowired
    private ITOrderExtService itOrderExtService;
    /**
     * 顺序消费并发级别
     */
    private Integer concurrent;

    /**
     * order业务顺序消费池
     */
    private KafkaConsumerPool<OrderDTO> kafkaConsumerPool;

    /**
     * 初始化顺序消费池
     */
    @PostConstruct
    public void init(){
        KafkaSortConsumerConfig<OrderDTO> config = new KafkaSortConsumerConfig<>();
        config.setBizName("order");
        config.setBizService(orderService::solveRetry);
        config.setConcurrentSize(concurrent);
        kafkaConsumerPool = new KafkaConsumerPool<>(config);
    }

    @KafkaListener(topics = {"${kafka.order.topic}"}, containerFactory = "baiyanCommonFactory")
    public void consumerMsg(List<ConsumerRecord<?, ?>> records, Acknowledgment ack){
        if(records.isEmpty()){
            return;
        }

        log.info("收到bigData推送的数据，拉取数据量：{}", records.size());
       // log.info("consumerRecord.value().toString()",  );
        records.forEach(consumerRecord->{
            OrderDTO order = GsonUtil.gsonToBean(consumerRecord.value().toString(), OrderDTO.class);
            log.info("kafka数据:{}",order);
            TOrderExt tOrderExt=new TOrderExt();
            BeanUtils.copyProperties(order,tOrderExt);
            tOrderExt.setOrderNo(order.getId()+"");
            LambdaQueryWrapper<TOrderExt> lambdaQueryWrapper=new LambdaQueryWrapper<TOrderExt>();
            lambdaQueryWrapper.eq(TOrderExt::getOrderNo,tOrderExt.getOrderNo()).last(" LIMIT 1");
            TOrderExt orderExt = itOrderExtService.getOne(lambdaQueryWrapper);
            if (ObjectUtils.isNotNull(orderExt)) {
                log.info("订单{}已存在,进行更新",tOrderExt.getOrderNo());
                LambdaUpdateWrapper<TOrderExt> updateWrapper=new LambdaUpdateWrapper<>();
                updateWrapper.eq(TOrderExt::getOrderNo,tOrderExt.getOrderNo());
                updateWrapper.set(TOrderExt::getOrderName,tOrderExt.getOrderName());
                updateWrapper.set(TOrderExt::getStatus,tOrderExt.getStatus());
                itOrderExtService.update(null,updateWrapper);
            } else {
                synchronized (this) {
                    if (ObjectUtils.isNotNull(orderExt)) {
                        log.info("订单{}已存在,进行更新",tOrderExt.getOrderNo());
                        LambdaUpdateWrapper<TOrderExt> updateWrapper=new LambdaUpdateWrapper<>();
                        updateWrapper.eq(TOrderExt::getOrderNo,tOrderExt.getOrderNo());
                        updateWrapper.set(TOrderExt::getOrderName,tOrderExt.getOrderName());
                        updateWrapper.set(TOrderExt::getStatus,tOrderExt.getStatus());
                        itOrderExtService.update(null,updateWrapper);
                    } else {
                        try {
                            log.info("订单{}不存在,进行保存",tOrderExt.getOrderNo());
                            itOrderExtService.save(tOrderExt);
                        } catch (Exception e) {
                            log.info("订单{}已存在,进行更新",tOrderExt.getOrderNo());
                            LambdaUpdateWrapper<TOrderExt> updateWrapper=new LambdaUpdateWrapper<>();
                            updateWrapper.eq(TOrderExt::getOrderNo,tOrderExt.getOrderNo());
                            updateWrapper.set(TOrderExt::getOrderName,tOrderExt.getOrderName());
                            updateWrapper.set(TOrderExt::getStatus,tOrderExt.getStatus());
                            itOrderExtService.update(null,updateWrapper);
                        }

                    }

                }
            }
          //  itOrderExtService.save(tOrderExt);
            kafkaConsumerPool.submitTask(order.getId(),order);

        });

        // 当线程池中任务处理完成的计数达到拉取到的记录数时提交
        // 注意这里如果存在部分业务阻塞时间很长，会导致位移提交不上去，务必做好一些熔断措施
        while (true){

           if(records.size() == kafkaConsumerPool.getPendingOffsets().get()){
               ack.acknowledge();
               log.info("offset提交：{}",records.get(records.size()-1).offset());
               kafkaConsumerPool.getPendingOffsets().set(0L);
               break;
           }
        }
    }




}
